package com.kele.test.rabbitmq.rpc;


import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

public class RPCClient {

    private static final String RPC_QUEUE = "rpc_queue";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("127.0.0.1");
        f.setUsername("guest");
        f.setPassword("guest");
        Connection connection = f.newConnection();
        Channel channel = connection.createChannel();
        /*
         * 定义队列 rpc_queue, 将从它接收请求信息
         *
         * 参数:
         * 1. queue, 对列名
         * 2. durable, 持久化
         * 3. exclusive, 排他
         * 4. autoDelete, 自动删除
         * 5. arguments, 其他参数属性
         */
        channel.queueDeclare(RPC_QUEUE,false,false,false,null);

        String msg = "3";
        //自动生成对列名,非持久,独占,自动删除
        String replyQueueName = channel.queueDeclare().getQueue();
        //生成关联id
        String corrId = UUID.randomUUID().toString();

        //设置两个参数:
        //1. 请求和响应的关联id
        //2. 传递响应数据的queue
        AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder();
        props.correlationId(corrId).replyTo(replyQueueName);

        //向 rpc_queue 队列发送请求数据, 请求第n个斐波那契数
        channel.basicPublish("", RPC_QUEUE, props.build(), msg.getBytes("UTF-8"));
        //接收响应数据的回调对象
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //如果响应消息的关联id,与请求的关联id相同,我们来处理这个响应数据
                if (message.getProperties().getCorrelationId().contentEquals(corrId)) {
                    //把收到的响应数据,放入阻塞集合
                    System.out.println(new String(message.getBody(), "UTF-8"));
                }
            }
        };

        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };

        //开始从队列接收响应数据
        channel.basicConsume(replyQueueName, true, deliverCallback, cancelCallback);

    }
}